Description

Problem: The notify_data_update method was declared as async but internally used synchronous zmq.Context and blocking zmq.Poller.poll() / socket.send_multipart() / socket.recv_multipart() calls, which blocked the asyncio event loop and degraded concurrency performance.

Solution:

• Replace the shared synchronous zmq.Context with zmq.asyncio.Context for the data update notification path.
• Use await sock.send_multipart() and await asyncio.wait_for(sock.recv_multipart(), ...) for truly non-blocking async I/O.
• Create per-call dynamic sockets in notify_data_update instead of reusing a long-lived data_status_update_socket, improving isolation and avoiding potential socket state corruption under concurrent calls.
• Optimize resource lifecycle: handshake socket and its synchronous context are now closed immediately after handshake completes, rather than being held for the entire object lifetime.

Changes

transfer_queue/storage/managers/base.py:

• Replace long-lived data_status_update_socket with per-call dynamic zmq.asyncio sockets in notify_data_update
• Replace zmq.Poller-based polling with asyncio.wait_for timeout mechanism